/**
 * Copyright (c) iwindplus Technologies Co., Ltd.2024-2030, All rights reserved.
 */

package com.iwindplus.boot.rabbit.service;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.text.CharSequenceUtil;
import com.iwindplus.boot.rabbit.domain.constant.RabbitConstant;
import com.iwindplus.boot.rabbit.domain.enums.RabbitExchangeTypeEnum;
import com.iwindplus.boot.rabbit.domain.property.RabbitProperty;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.CustomExchange;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.HeadersExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.beans.factory.SmartInitializingSingleton;

import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * rabbit动态注册.
 *
 * @author zengdegui
 * @since 2020/4/24
 */
@Slf4j
public class RabbitDynamicRegistry implements SmartInitializingSingleton {
    private AmqpAdmin amqpAdmin;

    private List<RabbitProperty.Config> configs;

    public RabbitDynamicRegistry(AmqpAdmin amqpAdmin, List<RabbitProperty.Config> configs) {
        this.amqpAdmin = amqpAdmin;
        this.configs = configs;
    }

    @Override
    public void afterSingletonsInstantiated() {
        log.info("RabbitMQ 根据配置动态创建和绑定队列、交换机");
        this.declareRabbitModule();
    }

    private void declareRabbitModule() {
        if (CollUtil.isEmpty(this.configs)) {
            return;
        }
        this.configs.forEach(config -> {
            //配置参数校验
            this.configParamValidate(config);
            // 队列
            Queue queue = this.convertQueue(config.getQueue());
            // 交换机
            Exchange exchange = this.convertExchange(config.getExchange());
            // 绑定关系
            String routingKey = config.getRoutingKey();
            Binding binding = BindingBuilder.bind(queue).to(exchange).with(routingKey).and(null);
            // 创建队列
            this.amqpAdmin.declareQueue(queue);
            // 创建交换机
            this.amqpAdmin.declareExchange(exchange);
            // 队列 绑定 交换机
            this.amqpAdmin.declareBinding(binding);
        });
    }

    private void configParamValidate(RabbitProperty.Config entity) {
        Assert.isTrue(CharSequenceUtil.isNotBlank(entity.getRoutingKey()), "RoutingKey is not null");
        Assert.isTrue(Objects.nonNull(entity.getExchange().getType()), "Exchange type is not null");
        Assert.isTrue(CharSequenceUtil.isNotBlank(entity.getExchange().getName()), "Exchange name is not null");
        Assert.isTrue(CharSequenceUtil.isNotBlank(entity.getQueue().getName()), "Queue name is not null");
    }

    private Queue convertQueue(RabbitProperty.Config.Queue queue) {
        Map<String, Object> arguments = queue.getArguments();
        if (MapUtil.isNotEmpty(arguments)) {
            // 转换ttl的类型为long
            final Object messageTtl = arguments.get(RabbitConstant.X_MESSAGE_TTL);
            if (Objects.nonNull(messageTtl)) {
                arguments.put(RabbitConstant.X_MESSAGE_TTL, Convert.toLong(messageTtl));
            }
            // 设置队列的优先级
            final Object maxPriority = arguments.get(RabbitConstant.X_MAX_PRIORITY);
            if (Objects.nonNull(maxPriority)) {
                arguments.put(RabbitConstant.X_MAX_PRIORITY, Convert.toLong(maxPriority));
            }
            // 是否需要绑定死信队列
            final Object deadLetterExchange = arguments.get(RabbitConstant.X_MAX_PRIORITY);
            final Object deadLetterRoutingKey = arguments.get(RabbitConstant.X_MAX_PRIORITY);
            if (Objects.isNull(deadLetterExchange) && Objects.isNull(deadLetterRoutingKey)
                    && CharSequenceUtil.isNotBlank(queue.getDeadLetterExchange())
                    && CharSequenceUtil.isNotBlank(queue.getDeadLetterRoutingKey())) {
                arguments.put(RabbitConstant.X_DEAD_LETTER_EXCHANGE, queue.getDeadLetterExchange());
                arguments.put(RabbitConstant.X_DEAD_LETTER_ROUTING_KEY, queue.getDeadLetterRoutingKey());
            }
        }
        return new Queue(queue.getName(), queue.isDurable(), queue.isExclusive(), queue.isAutoDelete(), arguments);
    }

    private Exchange convertExchange(RabbitProperty.Config.Exchange exchangeInfo) {
        RabbitExchangeTypeEnum exchangeType = exchangeInfo.getType();
        String exchangeName = exchangeInfo.getName();
        boolean isDurable = exchangeInfo.isDurable();
        boolean isAutoDelete = exchangeInfo.isAutoDelete();
        Map<String, Object> arguments = exchangeInfo.getArguments();
        if (RabbitExchangeTypeEnum.DIRECT.equals(exchangeType)) {
            return new DirectExchange(exchangeName, isDurable, isAutoDelete, arguments);
        } else if (RabbitExchangeTypeEnum.TOPIC.equals(exchangeType)) {
            return new TopicExchange(exchangeName, isDurable, isAutoDelete, arguments);
        } else if (RabbitExchangeTypeEnum.FANOUT.equals(exchangeType)) {
            return new FanoutExchange(exchangeName, isDurable, isAutoDelete, arguments);
        } else if (RabbitExchangeTypeEnum.HEADERS.equals(exchangeType)) {
            return new HeadersExchange(exchangeName, isDurable, isAutoDelete, arguments);
        } else if (RabbitExchangeTypeEnum.DELAYED.equals(exchangeType)) {
            arguments.put(RabbitConstant.X_DELAYED_TYPE, RabbitExchangeTypeEnum.DIRECT.getValue());
            return new CustomExchange(exchangeName, RabbitConstant.X_DELAYED_MESSAGE, isDurable, isAutoDelete, arguments);
        }
        return null;
    }
}
